package kn.pulldata.mq;


import cn.hutool.crypto.digest.DigestUtil;
import cn.hutool.http.HttpStatus;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.rabbitmq.client.Channel;
import kn.pulldata.common.OkHttpUtils;
import kn.pulldata.common.RabbitMqCon;
import kn.pulldata.common.ThreadPoolUtil;
import kn.pulldata.req.QueryTonalStateReq;
import kn.pulldata.resp.MatchDistanceResp;
import kn.pulldata.resp.PlanManagesItem;
import kn.pulldata.service.ExternalService;
import kn.pulldata.service.RedisLockService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

@Component
@Slf4j
public class PushConsumer {

    @Value("${monitor.keyword.saveUrl}")
    private String getKeywordSaveUrl;

    @Autowired
    private ExternalService externalService;

    @Autowired
    private RedisLockService redisLockService;


    //@RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE,containerFactory = "factory1")
    public void keywordReceiveMessage(String message, Channel channel, Message rabbitMessage) {

        try {
            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
            if(matchDistanceRespList.isEmpty()){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }

            push(matchDistanceRespList);

            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);

        }
        catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
    @RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE0,containerFactory = "factory1")
    public void keywordReceiveMessage0(String message, Channel channel, Message rabbitMessage) {

        try {
            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
            if(matchDistanceRespList.isEmpty()){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            push(matchDistanceRespList);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);

        }
        catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE1,containerFactory = "factory1")
    public void keywordReceiveMessage1(String message, Channel channel, Message rabbitMessage) {

        try {
            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
            if(matchDistanceRespList.isEmpty()){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            push(matchDistanceRespList);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);

        }
        catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }


    @RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE2,containerFactory = "factory1")
    public void keywordReceiveMessage2(String message, Channel channel, Message rabbitMessage) {

        try {
            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
            if(matchDistanceRespList.isEmpty()){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            push(matchDistanceRespList);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);

        }
        catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE3,containerFactory = "factory1")
    public void keywordReceiveMessage3(String message, Channel channel, Message rabbitMessage) {

        try {
            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
            if(matchDistanceRespList.isEmpty()){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            push(matchDistanceRespList);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);

        }
        catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE4,containerFactory = "factory1")
    public void keywordReceiveMessage4(String message, Channel channel, Message rabbitMessage) {

        try {
            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
            if(matchDistanceRespList.isEmpty()){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            push(matchDistanceRespList);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);

        }
        catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
    @Autowired
    private RabbitTemplate rabbitTemplate;

    private void push(List<MatchDistanceResp> matchDistanceRespList) {
        //List<String> urls = tonalStateConfig.getUrls();

        //UrlRotator urlRotator = new UrlRotator(urls);
        String url = "http://10.1.224.40/nlp/sentiment_classify";

        // 处理单个请求的逻辑
        long l = System.currentTimeMillis();
        List<CompletableFuture<MatchDistanceResp>> futures = matchDistanceRespList.stream()
                .map(resp -> CompletableFuture.supplyAsync(() -> singleRequest(resp,url), ThreadPoolUtil.getExecutor()))
                .collect(Collectors.toList());
        CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        allOf.thenRun(() -> {
            List<MatchDistanceResp> resps = futures.stream()
                    .map(CompletableFuture::join)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());

            if (!resps.isEmpty()) {
                String req = JSON.toJSONString(resps, SerializerFeature.WriteNullNumberAsZero, SerializerFeature.WriteNullListAsEmpty);
                long start = System.currentTimeMillis();
                if (pullData(req, getKeywordSaveUrl)) {
                    log.info("推送数据成功");
                }else {
                    log.error("推送数据重试次数超过后异常：{}",JSONUtil.toJsonStr(req));
                }
                long end = System.currentTimeMillis();
                //log.info("推送数据耗时：{}毫秒",(end - start));
            }
        });

    }


    private MatchDistanceResp singleRequest(MatchDistanceResp resp, String nextUrl) {
        //去重
        try {
            List<PlanManagesItem> planManagesItemList = new ArrayList<>();

            List<String> keyList = new ArrayList<>();
            for (PlanManagesItem planManage : resp.getPlanManages()) {
                String key = planManage.getPlanManageId() + resp.getWebsite();
                String uniqueKeyHash = DigestUtil.md5Hex16(key);
                keyList.add(uniqueKeyHash);
            }
            long start = System.currentTimeMillis();

            List<Boolean> lockResults = redisLockService.obtainLocksInBatch(keyList, 3600 * 24);
            long end = System.currentTimeMillis();
            //log.info("上锁耗时：{}毫秒",(end - start));
            for (int i = 0; i < lockResults.size(); i++) {
                if (lockResults.get(i)) {
                    planManagesItemList.add(resp.getPlanManages().get(i));
                }
            }
            if(planManagesItemList.isEmpty()){
                return null;
            }
            resp.setPlanManages(planManagesItemList);
        }catch (Exception e){
            e.printStackTrace();
            log.error("去重出错,异常信息{}", e.getMessage());
        }
        QueryTonalStateReq queryTonalStateReq = new QueryTonalStateReq();
        queryTonalStateReq.setTitle(resp.getTitle());
        queryTonalStateReq.setContent(resp.getContent() + " " + resp.getImageContent());
        long start = System.currentTimeMillis();
        String tonalState = externalService.queryTonalState(queryTonalStateReq,nextUrl);
        long end = System.currentTimeMillis();
        //log.info("调性耗时：{}毫秒",(end - start));
        resp.setTonalState(tonalState);
        return resp;

//        List<Long> idList = resp.getPlanManages().stream().map(PlanManagesItem::getPlanManageId).collect(Collectors.toList());
//        // 西安地铁
//        Long xianSubwayMonId = 18888L;
//        // 沈阳应急管理局地铁
//        Long shenYangMonId = 19999L;
//        if(idList.contains(xianSubwayMonId)){
//            // 准备发送
//            String messageId = UUID.randomUUID().toString();
//            // **深拷贝 resp**
//            MatchDistanceResp sendResp = JSON.parseObject(JSON.toJSONString(resp), MatchDistanceResp.class);
//            // 过滤出西安地铁或沈阳应急管理局的监测方案
//            sendResp.setPlanManages(resp.getPlanManages().stream()
//                    .filter(p -> p.getPlanManageId().equals(xianSubwayMonId))
//                    .collect(Collectors.toList()));
//            // 只将西安地铁的监测方案留下并发送
//            sendResp.setIsXiAnSubway(1);
//            String jsonStr = JSONUtil.toJsonStr(sendResp);
//            Message message = MessageBuilder.withBody(jsonStr.getBytes(StandardCharsets.UTF_8))
//                    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
//                    .setMessageId(messageId)
//                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
//                    .build();
//            CorrelationData correlationData = new CorrelationData(messageId);
//            // 发送给西安进行消费
//            rabbitTemplate.send(RabbitMqCon.XIAN_MONITOR_EXCHANGE, RabbitMqCon.XIAN_SUBWAY_KEY, message, correlationData);
//            log.info("测试西安地铁机构AI调性数据推送成功：{}",(sendResp.getWebsite()));
//
//            // 过滤掉西安地铁和沈阳应急管理局的监测方案，其他照常给
//            List<PlanManagesItem> collect = resp.getPlanManages().stream()
//                    .filter(p -> !p.getPlanManageId().equals(xianSubwayMonId))
//                    .collect(Collectors.toList());
//            resp.setPlanManages(collect);
//        }
//
//        if(idList.contains(shenYangMonId)){
//            // 准备发送
//            String messageId = UUID.randomUUID().toString();
//            // **深拷贝 resp**
//            MatchDistanceResp sendResp = JSON.parseObject(JSON.toJSONString(resp), MatchDistanceResp.class);
//            // 过滤出沈阳应急管理局的监测方案
//            sendResp.setPlanManages(resp.getPlanManages().stream()
//                    .filter(p -> p.getPlanManageId().equals(shenYangMonId))
//                    .collect(Collectors.toList()));
//            // 只将=沈阳的监测方案留下并发送
//            sendResp.setIsShenYang(1);
//            String jsonStr = JSONUtil.toJsonStr(sendResp);
//            Message message = MessageBuilder.withBody(jsonStr.getBytes(StandardCharsets.UTF_8))
//                    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
//                    .setMessageId(messageId)
//                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
//                    .build();
//            CorrelationData correlationData = new CorrelationData(messageId);
//            // 发送给沈阳进行消费
//            rabbitTemplate.send(RabbitMqCon.XIAN_MONITOR_EXCHANGE, RabbitMqCon.SHENYANG_MANAGE_KEY, message, correlationData);
//            log.info("测试沈阳应急管理局机构AI调性数据推送成功：{}",(sendResp.getWebsite()));
//
//            // 过滤掉沈阳应急管理局的监测方案，其他照常给
//            List<PlanManagesItem> collect = resp.getPlanManages().stream()
//                    .filter(p -> !p.getPlanManageId().equals(shenYangMonId))
//                    .collect(Collectors.toList());
//            resp.setPlanManages(collect);
//        }
//        if(resp.getPlanManages().isEmpty()){
//            return null;
//        }
    }

    private boolean pullData(String req, String url){
        // 处理消息逻辑
        int maxRetries = 3; // 最大重试次数
        int attempts = 0; // 当前尝试次数
        boolean success = false;

        while (attempts < maxRetries) {
            try {
                String body = OkHttpUtils.postJson(url, req);
                cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(body);
                if (HttpStatus.HTTP_OK == jsonObject.getInt("code")) {
                    success = true; // 请求成功，退出重试
                    log.info("推送成功！！！！！！！！！");
                    return success;
                } else {
                    log.error("推送失败");
                    attempts++;

                    // 增加等待时间，避免频繁请求
                    try {
                        Thread.sleep(1000); // 等待1秒后重试
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt(); // 恢复中断状态
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException("推送数据异常:"+ e.getMessage());
            }

        }
        return false;
    }




    //**********************************测试推送队列**********************************//
    //@RabbitListener(queues = RabbitMqCon.KEYWORD_TEST_PUSH_QUEUE)
    public void testMessage(String message, Channel channel, Message rabbitMessage) {

        List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
        log.info("测试推送队列监听到消息:{},个数：{}" , rabbitMessage.getMessageProperties().getMessageId(),matchDistanceRespList.size());

        //keywordReceiveMessage(message, channel, rabbitMessage);

    }


}
